package com.lcy.demo.kafka;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by PC on 2017/4/1.
 */
public class StreamDemo {
    public static void main(String[] args){
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "jd_group");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());

        StreamsConfig config = new StreamsConfig(props);

        KStreamBuilder builder = new KStreamBuilder();

        KStream<String,String> kStream = builder.stream("topic1");
        kStream.mapValues(new ValueMapper<String, String>() {
            @Override
            public String apply(String value) {
                return value==null?"0":value.length()+"";
            }
        })
        .to("TEST-TOPIC");


        KafkaStreams streams = new KafkaStreams(builder, config);

        streams.start();

    }
}
